package com.example.mqtt.client;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.extern.slf4j.Slf4j;

import java.util.Arrays;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;

import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED;

@Slf4j
public class MqttClientApplication {

    public static ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();

    public static void main(String[] args) {


        IMqttClient client = new RetryMqttClient.Builder()
                .address("device1-api.10646.cn")
                .clientId("linxy_mqttfx")
                .username("linxy_mqttfx")
                .password("QWn12afNPhwil4dP")
                .cleanSession(false)
//                .keepAliveTimeSeconds(10)
                .eventCallback(new RetryMqttClient.EventCallback() {
                    @Override
                    public void onLoginResult(RetryMqttClient c, MqttConnectReturnCode code) {

                        if(code == CONNECTION_ACCEPTED) {
                            log.info("connect to mqtt broker success");
                            Arrays.asList("thing/linxy_mqttfx/attribute/#", "thing_disconnect", "reply/#").forEach(topic -> {
                                c.subscribe(topic).whenComplete(new BiConsumer<Void, Throwable>() {
                                    @Override
                                    public void accept(Void aVoid, Throwable e) {
                                        if(e != null) {
                                            log.error("sub to topic {} error", topic, e);
                                        } else {
                                            log.info("sub top topic {} success", topic);
                                        }
                                    }
                                });
                            });

                            ByteBuf buf = Unpooled.buffer();
                            buf.writeBytes("{\"cmd\":{\"command\":\"attribute.publish\",\"params\": {\"thingKey\":\"linxy_mqttfx\",\"key\":\"server_req\",\"value\":\"{\\\"cmd\\\":\\\"controlLight\\\",\\\"seq\\\":1,\\\"data\\\":{\\\"light\\\":1,\\\"duration\\\":30,\\\"rate\\\":30}}\"}}}".getBytes());

                            c.publish("api", buf, MqttQoS.AT_LEAST_ONCE, false).whenComplete((aVoid, throwable) -> {

                            });

                            service.scheduleAtFixedRate(() -> {
                                c.close2();
                            }, 5, 5, TimeUnit.SECONDS);


                        } else {
                            log.error("connect to mqtt broker error, return code: {}", code);
                        }
                    }

                    @Override
                    public void onMessage(MqttPublishMessage msg) {
                        log.info("recv message: {}", msg);
                    }
                })
                .build();

        client.connect();





    }
}
